package Sun

import java.util.Date

import org.apache.spark.SparkContext
import org.apache.spark.mllib.linalg._
import org.apache.spark.mllib.regression.LabeledPoint
import org.apache.spark.rdd.RDD
import org.apache.spark.sql.types.{StringType, StructField, StructType}
import org.apache.spark.sql.{DataFrame, Row}
import org.apache.spark.storage.StorageLevel
import redis.clients.jedis.{Jedis, Response}

import scala.Array._
import scala.collection.mutable.ArrayBuffer


/**
 * Created by SHANGMAI on 2016/7/15.
 */

//对于不同feature进行打散


object Separate {

  //  x_{e}, x_{d}
  //  0	Mzid,
  //  1	interval,
  //  2,3,4	action type(click,impresion, other actions),
  //  (5,m+5)	website(separated by ,),
  //  (m+6,m+n+6)	type(separated by ,),
  //  m+n+7	#pre-clicks,
  //  m+n+8	#pre-impressions,
  //  m+n+9	#pre total actions,
  //  (m+n+10,2m+n+10)	website of pre-actions(separated by ,),
  //  (2m+n+11,2m+2n+11)	type of pre-actions(separated by ,)

  //从df提取出向量的函数,做中结果输出RDD 转换成标准结果

  def toVecotr(admonitorJoin: DataFrame): Unit = {

    //第一步 需要将DF中的内容离散化  变成浮点结构,同时要区分是否有转换
    //给出一个返回temp的rdd

    val now = new Date()

    var tempRdd = admonitorJoin.map(t =>

      //0	Mzid "mzid,eid,caid,spid,br,os,java,action,exposure_time,moring,afternoon,night,weekday"
      //最终得到结果是（uid.conversiontime.featureC),(chan,time,e,d....)
      (
      //key的三元组
      (t.getAs[String]("mzid"),timeConversion(t.getAs[String]("conversion_time")),t.getAs[String]("value")),

        //value
          {
           var arrayBufferResult = ArrayBuffer[Double]()

           //给出监测点信、渠道
           arrayBufferResult  += t.getAs[String]("spid").toDouble
          // 1 interval,
          //arrayBufferResult  += (now.getTime.toDouble - t.getAs[String]("exposure_time").toDouble)

           //time发生的时间
          arrayBufferResult  +=  t.getAs[String]("exposure_time").toDouble

           //  2,3,4	action type(click,impresion, other actions),
          arrayBufferResult  ++= actionSeparate(t.getAs[String]("action"))

          //  (5,m+5)	website(separated by ,)
          arrayBufferResult ++= websiteSeparate(t.getAs[String]("adweb"))

          // (m+6,m+n+6)	type(separated by ,),
          arrayBufferResult ++= typeSeparate(t.getAs[String]("adtype"))

          //arrayBufferResult.toArray

          //  m+n+7	#pre-clicks, //需要统计量
          //:+ preClicksSeparate(t.getAs[String]("adtype"))

          //  m+n+8	#pre-impressions, //需要统计的浏览数
          //:+ actionSeparate(t.getAs[String]("adtype"))
          arrayBufferResult.toArray
         }

        ) // K-V end

    )//end map

    //第二步  将离散化的数据进行reduce聚合 以user为维度聚合
   //.reduceByKey((x,y) => x ++= y )//进行按用户来聚合,将行为进行聚合
    .reduceByKey((x,y) => concat(x,y)) //进行按用户来聚合,将行为进行聚合

    //进行featureC的处理
    //进行website of pre-actions  type of pre-actions的处理

    .map(x => (x._1,
    {
       //开头给出mzid和转换时间
      //给出web的统计量 进行website of pre-actions  type of pre-actions的处理
      //val arrayChannel = pre_SPID_Action_Separate(x._1,sc)
      val arrayWeb = pre_ActionsSeparate(x._2,"web")
      val arraytype = pre_ActionsSeparate(x._2,"type")

      //featureC的的赋值
      //进行连接，给出featureC作
       concat(concat(arrayWeb,arraytype),x._2)
    }))

    // tempRdd.persist(StorageLevel.MEMORY_AND_DISK_SER)

    //给出两个集合的分割，根据转换时间的不同
    val sample_positive = tempRdd.filter(x => x._1._2 != "0")

    val sample_negative = tempRdd.filter(x => x._1._2 == "0")


    //给出一个文件存储的格式
    BasicFunction.saveSampleData(sample_positive, BasicFunction.sample_positive_save_path)
    BasicFunction.saveSampleData(sample_negative, BasicFunction.sample_negative_save_path)

    //返回一个规定类型函数
  }

  //将转化时间，如果没有转换时间则给0
  def timeConversion(conversion: String): String = {

    var result: String = null
    //当没有转换的时间的时候返回0
    if (conversion == "null")
      result = "0"

    //否则就返回转换时间本身
    else
      result = conversion

    return result
  }


  //给出trianing的格式
  def returnTraining(rdd: RDD[((String, Double), SparseVector)]): RDD[LabeledPoint] = {
    rdd.map(x => LabeledPoint(x._1._2, x._2))
  }

  //给出 predict格式
  def reutrnPredict(): Unit = {

  }

  //返回了featureC的相关参数。给出了featureC的相关列表，给出了user在channel上的点击的统计量
  def pre_SPID_Action_Separate(mzid: String, sc: SparkContext): Array[Double] = {


    val sqlContext = new org.apache.spark.sql.SQLContext(sc)

    val jedis = RedisClient.pool.getResource
    jedis.select(BasicFunction.dbIndex)

    var tempMap = Map[Int, Double]()
    //给出mid与channel的得分
    val finalResult = sqlContext.read.format("json").load(BasicFunction.result_compete_path)

    val arrayResult = finalResult.filter("mzid =" + mzid).take(100)


    for (i <- 0 to arrayResult.size) {
      //给出得到的渠道数
      val indexkey = "adchannel" + arrayResult.apply(i).apply(1)

      //如果key存在
      if (jedis.exists(indexkey)) {
        //给出channel的顺序
        val index = jedis.get(indexkey).toInt
        val value = arrayResult.apply(i).apply(2).toString.toDouble

        tempMap += (index -> value)
      }

      val count = jedis.get("adwebcount").toInt
    }
    //关闭释放内存

    jedis.close()

    //给出结果
    arrayCountReturn(BasicFunction.channelSize, tempMap).toArray
  }

  def pre_ActionsSeparate(feature: Array[Double], kind: String): Array[Double] = {
    val jedis = RedisClient.pool.getResource
    jedis.select(BasicFunction.dbIndex)

    //统计个数的的字典
    var tempMap = Map[Int, Double]()

    //给出两个的总个数
    val adtypecount = jedis.get("adtypecount").toInt
    val adwebcount = jedis.get("adwebcount").toInt

    var kindname: String = null

    //特征开始的技术位置
    var start = 0

    //特征的个数
    var kindCount = 0

    //特征距离action的间隔
    var actionInterval = 0

    //根据类型不同来区分不同的名称，目前有两种 type web给出名字和开始位置
    if (kind == "web") {
      kindname = "adweb"
      start = 0
      kindCount = adwebcount
      actionInterval = 10
    }
    else {
      kindname = "adtype"
      start = 0 + adwebcount
      kindCount = adtypecount
      actionInterval = 12
    }

    //val start = jedis.get(kindname + "count").to

    //遍历array 寻找渠道 如果转化 则标记为1.0 否则为0
    for (i <- range(0: Int, feature.size - 1: Int, BasicFunction.featureD_length + BasicFunction.featureE_length + 2: Int)) {

      for (index <- 0 to kindCount) {

        //当是在类型/网站（等） 有行为的时候 的时候
        val nowindex = i + start + index
        if (feature.apply(nowindex) == 1.0) {

          //当没有存储的时候，赋初值
          if (tempMap.contains(index) == false) {
            tempMap += (index -> 0.0)
          }
          //如果是点击的时候 则加5
          if (feature.apply(i + 2) == 1)
            tempMap += (index -> (tempMap(index) + 5.0))
          //如果是浏览 则加1
          else if (feature.apply(i + 3) == 1)
            tempMap += (index -> (tempMap(index) + 1.0))
        }
      }
    }
    //关闭释放内存
    jedis.close()

    //给出结果
    arrayCountReturn(kindCount, tempMap).toArray
  }


  //进行动作字段的切割
  def actionSeparate(action: String): ArrayBuffer[Double] = {
    //次数数量较少 直接判断

    var result = ArrayBuffer[Double]()
    //当为点击的时候
    if (action == "1") {
      val tempList = List[Int](1)
      result = arrayReturn(3, tempList)
    }
    else if (action == "0") {
      val tempList = List[Int](2)
      result = arrayReturn(3, tempList)
    }
    //当为其他的数值的时候
    else {
      val tempList = List[Int](3)
      result = arrayReturn(3, tempList)
    }

    return result
  }

  //web的打散
  def websiteSeparate(web: String): ArrayBuffer[Double] = {
    val jedis = RedisClient.pool.getResource

    var tempList = List[Int]()
    jedis.select(BasicFunction.dbIndex)
    val value = jedis.get("adweb" + web)
    //判断有无匹配
    if (value != null) {
      tempList :+= (value.toInt + 1)
    }

    val count = jedis.get("adwebcount").toInt
    //关闭释放内存

    jedis.close()
    //添加web类型
    arrayReturn(count, tempList)
  }

  def typeSeparate(adtype: String): ArrayBuffer[Double] = {

    val jedis = RedisClient.pool.getResource

    var tempList = List[Int]()
    jedis.select(BasicFunction.dbIndex)

    //给出“+” 分割符的的广告类型
    var onetype = Array[String](adtype)
    if (adtype.contains("+")) {
      onetype = adtype.split("\\+")
    }

    for (i <- 0 to onetype.length - 1) {
      val value = jedis.get("adtype" + onetype.apply(i))
      if (value != null) {
        tempList :+= (value.toInt + 1)
      }
    }

    val count = jedis.get("adtypecount").toInt
    //关闭释放内存
    jedis.close()

    arrayReturn(count, tempList)
  }


  //一个进行数组返回的函数
  //第一个是数组的长度，第二参数是为1值的位置
  def arrayReturn(size: Int, trueValue: List[Int]): ArrayBuffer[Double] = {
    var result = ArrayBuffer[Double]()

    //有的含有两个字符，分别打散
    for (i <- 1 to size) {
      if (trueValue.contains(i))
        result = result :+ 1.0;
      else
        result = result :+ 0.0;
    }
    return result
  }

  //一个进行统计量（总数，个数）返回的函数
  //第一个是数组的长度，第二参数是为1值的位置
  def arrayCountReturn(size: Int, count: Map[Int, Double]): ArrayBuffer[Double] = {
    var result = ArrayBuffer[Double]()

    //有的含有两个字符，分别打散
    for (i <- 1 to size) {

      //如果存在这个KEY
      if (count.contains(i))
        result = result :+ count(i)

      //没有的时候，赋予0
      else
        result = result :+ 0.0

    }
    return result
  }

  def computeList(channelStatics: DataFrame) {

    //用该模型整理所有的渠道
    val spidDf = channelStatics.select("spid").distinct

    //用RDD方式的存储
    spidDf.map(t => t.getAs[String]("spid")).repartition(1).saveAsTextFile(BasicFunction.channelList_path)
  }

  //给出计算FeatureC的公式，目前就是计算每个用户在每个渠道上点击的次数

  def compute_statistics_FeatureC(channelStatics: DataFrame, sc: SparkContext): Unit = {

    //得到某个人在某个渠道上的动作
    val channelGroup = channelStatics.groupBy("mzid", "spid", "action")
      .agg(Map("exposure_time" -> "count"))
      .toDF("mzid", "spid", "action", "count")
      .map(x => (x.getAs[String]("mzid"),x.getAs[String]("spid") + "," + {
        //根据action来进行赋值
        var value:Long = 0
        if  (x.getAs[String]("action") == "1")
          value = x.getAs[Long]("count") * 5
        else
          value = x.getAs[Long]("count")

        value.toString
      }
      ))

      //进行相加
      .reduceByKey((x,y) => x + "," + y)
 
      //进行最终转换到df 用于存储
      .mapPartitions(partition => {

         //按区块来进行相加，减少redis读写次数

         val er = 1
         val edrf = 2
         var jedis:Jedis = null

         jedis = RedisClient.pool.getResource
         jedis.select(BasicFunction.dbIndex)
         val pp = jedis.pipelined()
         var temptest:Response[String] = null
         var couunt = 1
         val partitionResult1 = partition.toList.map(x => (x._1, {
         //给出一个存储redis response结果的类型
         var tempRedisRes = Map[Response[String],Double]()

         //等于将将value串分割
         val todeal = x._2.split(",")

         for(i <- 0 to todeal.size - 1){

            //在i是偶数的时候
            if(i % 2 == 0)
            {
              val indexkey = "adchannel" + todeal.apply(i)
              val index1 = pp.get(indexkey)

              //给出channel的顺序
              val value = todeal.apply(i+1).toDouble

              //记录reponse与数值的关系
              tempRedisRes = tempRedisRes + (index1 -> value)
            }
          }

          tempRedisRes
          }
          )) //经历集体取数阶段 将（id，(channel1,value1,channel2,value2....) 变成中间值 （id，Map[Response[String],Double]）
            //进行取数

         pp.sync()

         //var partitionResult:Iterator[(String,String)] = null

         val partitionResult =  partitionResult1.map(x => (x._1,{

            //统计个数的的字典
            var tempMap = Map[Int, Double]()
            //对每个response进行取数
            x._2.keys.foreach{key => {
               val index = key.get()
               //如果key存在
               if (index != null) {
                  tempMap += (index.toInt -> x._2(key))
                  }
              }
            }

        //进行向量转换
        var result = arrayCountReturn(BasicFunction.channelSize, tempMap).toString
        result = result.replace("ArrayBuffer(","").replace(")","")
        result
        })).toIterator
        // end try


      //重写一个没有观的澳的
//         partitionResult = partition .map(x => (x._1,{
//
//            //统计个数的的字典
//         var tempMap = Map[Int, Double]()
//
//         var tempRedisRes = Map[Response[String],Double]()
//
//         //等于将将value串分割
//         val todeal = x._2.split(",")
//
//         for(i <- 0 to todeal.size - 1){
//
//            //在i是偶数的时候
//            if(i % 2 == 0)
//            {
//              val indexkey = "adchannel" + todeal.apply(i)
//              val index = jedis.get(indexkey)
//
//              //给出channel的顺序
//              val value = todeal.apply(i+1).toDouble
//
//              //记录reponse与数值的关系
//              if (index != null) {
//                tempMap = tempMap + (index.toInt -> value)
//            }
//          }
//          }
//
//        //进行向量转换
//        var result = arrayCountReturn(BasicFunction.channelSize, tempMap).toString
//        result = result.replace("ArrayBuffer(","").replace(")","")
//        result
//        }))
//        } // end try

//       catch {
//            case e: Exception => {
//            println("Redis-Timeout" + e)
//       }}
//      finally{

       jedis.close()
       partitionResult
       }
       )

     //转化成Row
    .map(x => Row(x._1,x._2))

    var fields = List[StructField]()
    fields = fields :+ StructField("mzid2", StringType, true)
    fields = fields :+ StructField("value", StringType, true)

    val structType_fieldsr = StructType.apply(fields)

    val sqlContext = new org.apache.spark.sql.SQLContext(sc)
    var df_Channel = sqlContext.createDataFrame(channelGroup, structType_fieldsr)


    //进行df的存储,为了进行存储写入而使用
    df_Channel.write.format("json").save(BasicFunction.featureC_path)
    //进行相加
    //channelStatics.repartition(1).write.format("json").save("H:/MutiTouch/ChannelStatics")

  }

}